package com.atzc.app;


import com.atzc.bean.CarDetailLogBean;
import com.atzc.bean.CarListLogBean;
import com.atzc.bean.HomeLogBean;
import com.atzc.bean.StartLogBean;
import com.atzc.sink.CarDetailLogSinkMysql;
import com.atzc.sink.CarListLogSinkMysql;
import com.atzc.sink.HomeLogSinkMysql;
import com.atzc.source.KafkaSourceUtils;
import com.atzc.utils.filter.ContainsDoubleOrFilterFunc;
import com.atzc.utils.filter.ContainsFilterFunc;
import com.atzc.utils.map.ParseCarDetailLogBeanMap;
import com.atzc.utils.map.ParseCarListLogBeanMap;
import com.atzc.utils.map.ParseHomeLogBeanMap;
import com.atzc.utils.map.ParseStartLogBeanMap;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;



public class FlinkAPP {
    public static void main(String[] args) throws Exception {
        // 获取环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 写上主题  获取kafka 数据源  设置并行度
        String topic = "test";
        int parallelism = 1;
        DataStreamSource<String> streamSource = env.addSource(KafkaSourceUtils.getFlinkKafkaConsumer010(topic)).setParallelism(parallelism);


        // 1.车辆详情页日志数据  可以是两个关键字
        // 过滤数据分流
        SingleOutputStreamOperator<String> carDetailLogSingleStream = streamSource.filter(new ContainsDoubleOrFilterFunc("/car/detail/renter reqParm","/longRentCar/detail/renter response"));

        // 切割字符串转换为bean
        SingleOutputStreamOperator<CarDetailLogBean> beanSingleOutputStreamOperator = carDetailLogSingleStream.map(new ParseCarDetailLogBeanMap());


        // 删除错误格式的日志
        SingleOutputStreamOperator<CarDetailLogBean> filterCarDetailLogBeanSingleOutputStreamOperator = beanSingleOutputStreamOperator.filter(new FilterFunction<CarDetailLogBean>() {
            @Override
            public boolean filter(CarDetailLogBean carDetailLogBean) throws Exception {

                return carDetailLogBean != null;
            }
        });

        // 输出到mysql中
        filterCarDetailLogBeanSingleOutputStreamOperator.addSink(new CarDetailLogSinkMysql());

        // 2. 首页日志处理
        // 过滤数据分流
        SingleOutputStreamOperator<String> homeLogSingleStream = streamSource.filter(new ContainsFilterFunc("/homepage/homeData response"));

        // 切割字符串转换为bean
        SingleOutputStreamOperator<HomeLogBean> homeLogBeanSingleOutputStreamOperator = homeLogSingleStream.map(new ParseHomeLogBeanMap());

        // 删除错误格式的日志
        SingleOutputStreamOperator<HomeLogBean> filterHomeLogBeanSingleOutputStreamOperator = homeLogBeanSingleOutputStreamOperator.filter(new FilterFunction<HomeLogBean>() {
            @Override
            public boolean filter(HomeLogBean homeLogBean) throws Exception {

                return homeLogBean != null;
            }
        });

        // 输出到mysql
        filterHomeLogBeanSingleOutputStreamOperator.addSink(new HomeLogSinkMysql());


        // 3. 启动日志处理
        // 过滤数据分流
        SingleOutputStreamOperator<String> startLogSingleStream = streamSource.filter(new ContainsFilterFunc("/screenPic/getPic response"));

        // 切割字符串转换为bean
        SingleOutputStreamOperator<StartLogBean> startLogBeanSingleOutputStreamOperator = startLogSingleStream.map(new ParseStartLogBeanMap());

        // 删除错误格式的日志
        SingleOutputStreamOperator<StartLogBean> filterStartLogBeanSingleOutputStreamOperator = startLogBeanSingleOutputStreamOperator.filter(new FilterFunction<StartLogBean>() {
            @Override
            public boolean filter(StartLogBean startLogBean) throws Exception {

                return startLogBean != null;
            }
        });

        // 4. 列表页  包含长租车辆列表和短租车辆列表
        // 过滤数据分流
        SingleOutputStreamOperator<String> carListLogSingleStream = streamSource.filter(new ContainsDoubleOrFilterFunc("/longRentCar/search/car/list response","/search/car/list response"));

        // 切割字符串转换为bean
        SingleOutputStreamOperator<CarListLogBean> carListLogBeanSingleOutputStreamOperator = carListLogSingleStream.map(new ParseCarListLogBeanMap());

        // 删除错误格式的日志
        SingleOutputStreamOperator<CarListLogBean> filterCarListLogBeanSingleOutputStreamOperator = carListLogBeanSingleOutputStreamOperator.filter(new FilterFunction<CarListLogBean>() {
            @Override
            public boolean filter(CarListLogBean carListLogBean) throws Exception {

                return carListLogBean != null;
            }
        });

        // 输出到mysql
        filterCarListLogBeanSingleOutputStreamOperator.addSink(new CarListLogSinkMysql());


        // TODO 添加缓存和状态后端
        // 执行
        env.execute();
    }
}
